package com.esdc.paymentsystem;

import com.esdc.paymentsystem.exception.MemberNotExistException;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Resource;
import javax.ejb.Stateful;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

/**
 *
 * @author shou1dwe
 */
@Stateful
public class PaymentCoordinatorBean implements PaymentCoordinatorRemote {
    private static final int RETRIEVE_OUTSTANDING_FINE = 0;
    private static final int PAY_FINE = 1;

    @Resource(mappedName = "jms/testqfc")
    private ConnectionFactory queueConnectionFactory;
    @Resource(mappedName = "jms/PaymentQueue")
    private Queue queue;
    
    private MapMessage response = null;
    private final Object waitUntilDone = new Object();

    public Double getOutstandingFine(Long memberId) throws MemberNotExistException {
        print("getOutstandingFine request from client received.");
        final Map<String, Object> messageContentMap = new HashMap<String, Object>();
        messageContentMap.put("RequestType", RETRIEVE_OUTSTANDING_FINE);
        messageContentMap.put("Id", memberId);
        return sendSyncRequest(memberId, messageContentMap, "Fine");
    }

    public Double payFine(Long memberId, Double amount) throws MemberNotExistException {
        print("payFine request from client received.");
        final Map<String, Object> messageContentMap = new HashMap<String, Object>();
        messageContentMap.put("RequestType", PAY_FINE);
        messageContentMap.put("Id", memberId);
        messageContentMap.put("Amount", amount);
        return sendSyncRequest(memberId, messageContentMap, "Fine");
    }

    private Double sendSyncRequest(Long memberId, final Map<String, Object> messageContentMap, String attribute) throws MemberNotExistException {
        final String messageId = String.format("%s%s", Long.toString(System.currentTimeMillis()), Long.toString(memberId));
        Thread t = new Thread(new Runnable() {
            public void run() {
                send(messageContentMap, messageId);
            }
        });
        t.start();
        try {
            synchronized (waitUntilDone) {
                waitUntilDone.wait(10000);
            }
            print("Return result to client");
            String error = response.getString("Error");
            if(error!=null){
                throw new MemberNotExistException(error);
            }
            return response.getDouble(attribute);
        } catch (JMSException ex) {
            Logger.getLogger(PaymentCoordinatorBean.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InterruptedException ex) {
            Logger.getLogger(PaymentCoordinatorBean.class.getName()).log(Level.SEVERE, null, ex);
        }
        
        return null;
    }

    private void send(Map contentMap, String messageId) {
        try {
            Connection queueConnection = queueConnectionFactory.createConnection();
            Session session = queueConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            MessageProducer producer = session.createProducer(queue);
            MapMessage message = session.createMapMessage();
            Queue tempQueue = session.createTemporaryQueue();
            message = transform(message, contentMap);
            message.setJMSReplyTo(tempQueue);
            message.setJMSCorrelationID(messageId);

            MessageConsumer consumer = session.createConsumer(tempQueue);
            consumer.setMessageListener(new PaymentPrivateListener());
            queueConnection.start();

            producer.send(message);
            print("correlationID: " + message.getJMSCorrelationID());
        } catch (JMSException ex) {
            Logger.getLogger(PaymentCoordinatorBean.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    private MapMessage transform(MapMessage message, Map contentMap) throws JMSException {
        for (Entry<String, Object> entry : (Set<Entry<String, Object>>) contentMap.entrySet()) {
            message.setObject(entry.getKey(), entry.getValue());
        }
        return message;
    }

    private void print(String msg) {
        System.out.println(String.format("[LPS] %s", msg));
    }

    private class PaymentPrivateListener implements MessageListener {
        public void onMessage(Message message) {
            print(":listener: Reply received");
            if (message instanceof MapMessage) {
                try {
                    print("JMSCorrelationID: " + message.getJMSCorrelationID());
                    print("MessageMessageID: " + message.getJMSMessageID());
                    response = (MapMessage) message;
                    synchronized (waitUntilDone) {
                        waitUntilDone.notify();
                    }
                } catch (JMSException ex) {
                    Logger.getLogger(PaymentCoordinatorBean.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    }
}
